package com.example.demo.controller;

import com.example.demo.common.R;
import org.springframework.data.mongodb.core.ReactiveMongoTemplate;
import org.springframework.data.mongodb.core.aggregation.Aggregation;
import org.springframework.http.codec.multipart.FilePart;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.stream.Collectors;

/**
 * @author fanhb on 2020/8/7
 * @desc xxxx
 */
public class ReactiveMongoController {

    @Resource
    private ReactiveMongoTemplate reactiveMongoTemplate;

    /**
     *
     * 文件上传
     * 解析CSV文件,然后插入到mongodb
     */
    public Mono<R> insertAll(FilePart filePart) {
        Flux<String> stringFlux = filePart.content().map(o -> o.toString(StandardCharsets.UTF_8))
                .collect(Collectors.joining())
                .map(o -> o.split("\\n"))
                .flatMapMany(Flux::fromArray);

        Flux<ChannelVO> map = stringFlux.map(line -> {
            String[] split = line.split(",");
            ChannelVO channelVO = new ChannelVO();
            channelVO.setId(split[0]);
            channelVO.setName(split[1]);
            return channelVO;
        });


//        Aggregation.unwind()
//        reactiveMongoTemplate.aggregate()

//        事务如何处理?????????
//        reactiveMongoTemplate.inTransaction().
        Flux<ChannelVO> channel = reactiveMongoTemplate.insertAll(map.collectList(), "Channel");
//        channel.subscribe(System.out::println);
        //返回这个flux channel有什么用呢
        // TODO: 2020/8/7
        return R.ok().body(channel);

    }
}
